package demo_day4;

import org.apache.commons.lang3.math.NumberUtils;
import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.record.Record;

import java.util.Map;

/*
    拦截器是 kafka 自己调用的，生产者用的
*/
public class KafkaIntercept implements ProducerInterceptor {

    @Override
    public void configure(Map<String, ?> configs) {
        // 初始化配置
        // 在最开始的时候运行一次
        System.out.println("配置初始化，只运行一次");
    }

    @Override
    public ProducerRecord onSend(ProducerRecord record) {
        // 拦截器中最主要的一个方法
        // 需求：通过拦截器，对数据中的工资 +2000 然后往后输出
        String[] arr = record.value().toString().split(",");
        int i = 0;
        if (NumberUtils.isDigits(arr[1])){
            i = Integer.parseInt(arr[1]) + 2000;
        }
        // record对象不能修改其中的任何属性，我们还想输出数据
        ProducerRecord retRecord = new ProducerRecord<>(
                record.topic(),
                record.partition(),
                record.key(),
                arr[0] + "," + i
        );
        return retRecord;
    }

    @Override
    public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
        // 这个就是ack应答时会调用，不写也没有任何关系
        // 发送几条数据就运行几次
        System.out.println("每一次ack应答，就会运行一次onAcknowledgement:" + metadata.toString());
    }

    @Override
    public void close() {
        // 处理完数据之后，拦截器会运行一次，整个发送，只会运行一次
        System.out.println("结束运行方法，只运行一次");
    }
}
